/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "xdrbase.h"
#include "channel.h"
#include "api.h"

static int record_keyvalue(FILE *, struct record *);
static struct record *get_jobrecord_fromfile(FILE *, unsigned int, JOBID_T *);
static int get_record_type(char *, int *);
static int read_keyvalue_rec(char *, struct record *);

#define   RECORD_TYPE_JOB_RELATED     1
#define   RECORD_TYPE_NON_JOB_RELATED 0

/**
 * @brief       get the record file handle according to the specified times.
 *
 * @param[in]   searchtime    #1: specified begin time and end time for searching
 * @param[out]  handle        #2: open the record file and set file handle
 *
 * @retval      0             #1: succeed
 * @retval      -1            #2: did not find the record file or failed to open it
 */
int
get_record_handle(time_t searchtime[], struct recordFileHandle *handle)
{
    char strWorkDir[PATH_MAX+1];
    snprintf(strWorkDir, PATH_MAX, "%s", baseParams[BASE_WORKDIR].string_value);

    int curOpenFile=-1, lastOpenFile=-1;
    bool_t findLast=FALSE, oldFormat=FALSE;

    char strRecordFile[PATH_MAX+1];
    unsigned int i;
    for (i = 1; ; i++) {

        snprintf(strRecordFile, PATH_MAX , "%s/qp.records.%d", strWorkDir, i);

        struct stat st;
        if (stat(strRecordFile, &st) == 0) {
            FILE *fileRecord;

            fileRecord = fopen(strRecordFile, "r");
            if (NULL == fileRecord) {
                log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "fopen", strRecordFile);
                return -1;
            }

            char ch;
            time_t logTime;
            if (fscanf(fileRecord, "%c%ld", &ch, &logTime) != 2 || ch != '#') {

                log_message(ERR, NOCLASS, "%s: fscanf(%s) failed: check the file format",
                            __func__, strRecordFile);
                fclose(fileRecord);
                oldFormat = TRUE;

                if (findLast == FALSE) {
                    lastOpenFile = i - 1;
                }
                curOpenFile = i - 1;
                break;
            }

            fclose(fileRecord);

            if (searchtime[1] >  logTime && !findLast) {
                findLast = TRUE;
                lastOpenFile = i - 1;
            }

            if (searchtime[0] < logTime) {
                continue;
            } else {
                curOpenFile = i--;
                break;
            }
        } else {
            if (findLast == FALSE) {
                lastOpenFile = i - 1;
            }

            curOpenFile = i - 1;
            break;
        }
    }

    if (oldFormat == TRUE) {
        if (curOpenFile == -1) {
            if (findLast == FALSE) {

                curOpenFile = 0;
                lastOpenFile = -1;
            } else {

                curOpenFile = i - 1;
            }
        }
    }

    if (curOpenFile == lastOpenFile) {
        lastOpenFile = -1;
    }

    if (0 > curOpenFile) {

        log_message(ERR, NOCLASS, "%s: current open record file number < 0", __func__);
        return -1;
    }

    if (curOpenFile == 0) {
        snprintf(strRecordFile, PATH_MAX, "%s/qp.records", strWorkDir);
    } else {
        snprintf(strRecordFile, PATH_MAX, "%s/qp.records.%d", strWorkDir, curOpenFile);
    }

    FILE *fileRecord;
    fileRecord = fopen(strRecordFile, "r");
    if (NULL == fileRecord) {
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "fopen", strRecordFile);
        return -1;
    }

    handle->file_handle = fileRecord;
    snprintf(handle->record_filename, PATH_MAX, "%s", strRecordFile);
    handle->current_fileno = curOpenFile;
    handle->last_fileno = lastOpenFile;

    return (0);
} // end function get_record_handle

/**
 * @brief       read one line of record from the opened file handle.
 *
 * @param[in|out]   handle    #1: specified begin time and end time for searching
 *
 * @retval      not NULL      #1: succeed
 * @retval      NULL          #2: failed to open one record file or reach to end of file
 *
 * @note        quickpool_errno is set to ERROR_EOF when reading to the end of the file.
 *              file_handle would be changed if reach to end of current file.
 */
struct record *
read_record_fromhandle(struct recordFileHandle *handle)
{
    struct record *logRec=NULL;
    unsigned int numLine=0;
    if (NULL != handle->file_handle) {
        logRec = read_file_line(handle->file_handle, &numLine);
    }

    if (NULL == logRec && handle->last_fileno >= 0
        && handle->current_fileno > handle->last_fileno) {

        if (NULL != handle->file_handle) {
            fclose(handle->file_handle);
        }

        char *strFile;
        strFile = strstr(handle->record_filename, "qp.records");
        if (NULL != strFile) {
            *(strFile-1) = '\0';
        }

        char strRecordFile[PATH_MAX+1];
        if (1 == handle->current_fileno) {

            snprintf(strRecordFile, PATH_MAX, "%s/qp.records", handle->record_filename);
            handle->current_fileno = 0;
            handle->last_fileno = -1;
        } else {
            snprintf(strRecordFile, PATH_MAX, "%s/qp.records.%d",
                     handle->record_filename, --(handle->current_fileno));
        }

        strncpy(handle->record_filename, strRecordFile, PATH_MAX);

        FILE *newFileHandle;
        newFileHandle = fopen(strRecordFile, "r");
        if (NULL == newFileHandle) {

            log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "fopen", strRecordFile);
            return NULL;
        }

        logRec = read_file_line(newFileHandle, &numLine);
        handle->file_handle = newFileHandle;
    }

    return logRec;
} // end function read_record_fromhandle

/**
 * @brief       read one line of the open file.
 *
 * @param[in]   filerecord    #1: the open file handler 
 * @param[out]  linenum       #2: the number of the reading line
 *
 * @retval      not NULL      #1: succeed
 * @retval      NULL          #2: failed
 *
 * @note        line format looks like: "type" "version" log_time num_kvs key "value" ...
 */
struct record *
read_file_line(FILE *filerecord, unsigned int *linenum)
{
    static struct record *logRec;

    if (NULL != logRec) {
        free_record(logRec);
    } else {

        logRec = (struct record *)calloc(1, sizeof(struct record));
        if (NULL == logRec) {

            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct record));
            quickpool_errno = ERROR_MEMORY;
            return NULL;
        }
    }

    quickpool_errno = 0;
    char buf[BUFSIZ];
    char *line;
    if (NULL == (line=gf_get_line(filerecord, linenum, buf))) {

        if (feof(filerecord)) {
            quickpool_errno = ERROR_EOF;
        }

        return NULL;
    }

    char *namebuf;
    namebuf = (char *)calloc(1, strlen(line)+1);
    if (NULL == namebuf) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", strlen(line)+1);
        quickpool_errno = ERROR_MEMORY;
        return NULL;
    }

    int ccount;
    ccount = strip_quotes(line, namebuf);
    if (0 > ccount || strlen(line) == (unsigned int)ccount
        || strlen(namebuf) >= MAXNAMELEN) {

        log_message(ERR, NOCLASS, "%s: recode line (%s) has format error, no record type found.",
                    __func__, line);
        quickpool_errno = ERROR_FILE_FORMAT;
        free(namebuf);

        return NULL;
    }

    char strRecType[MAXNAMELEN];
    snprintf(strRecType, MAXNAMELEN, "%s", namebuf);

    line += ccount + 1;
    ccount = strip_quotes(line, namebuf);
    if (0 > ccount || strlen(line) == (unsigned int)ccount
        || strlen(namebuf) >= MAXVERSIONLEN) {

        log_message(ERR, NOCLASS, "%s: record line (%s) has format error, no version found.",
                    __func__, line);
        quickpool_errno = ERROR_FILE_FORMAT;
        free(namebuf);

        return NULL;
    }

    if (atoi(namebuf) <= 0) {
        log_message(ERR, NOCLASS, "%s: record line (%s) has format error, wrong version <%s>.",
                    __func__, namebuf);
        quickpool_errno = ERROR_FILE_FORMAT;
        return NULL;
    }

    snprintf(logRec->version, MAXVERSIONLEN, "%s", namebuf);
    free(namebuf);

    line += ccount + 1;

    int cc;
    cc = sscanf(line, "%ld%n", &logRec->log_time, &ccount);
    if (1 != cc) {

        log_message(ERR, NOCLASS, "%s: record line (%s) has format error, no log time found.",
                    __func__, line);
        quickpool_errno = ERROR_FILE_FORMAT;
        return NULL;
    }

    line += ccount + 1;
    int kind;
    logRec->type = get_record_type(strRecType, &kind);
    if (0 > logRec->type) {

        log_message(ERR, NOCLASS, "%s: record line (%s) has format error, unknown record type <%s>.",
                    __func__, line, strRecType);
        return NULL;
    }

    quickpool_errno = read_keyvalue_rec(line, logRec);

    if (quickpool_errno == 0) {
        return logRec;
    }

    return NULL;
} // end function read_file_line

/**
 * @brief       free allocated memory for struct record.
 *
 * @param[out]  logrec        #1: record to be freed 
 *
 * @note        the function free the key values in record, not free the record itself.
 */
void
free_record(struct record *logrec)
{
    switch (logrec->type) {
    case RECORD_TYPE_JOB_NEW:
    case RECORD_TYPE_JOB_START:
    case RECORD_TYPE_PRE_EXEC_START:
    case RECORD_TYPE_JOB_EXECUTE:
    case RECORD_TYPE_JOB_START_ACCEPT:
    case RECORD_TYPE_JOB_STATUS:
    case RECORD_TYPE_JOB_FINISH:
    case RECORD_TYPE_JOB_MODIFY:
    case RECORD_TYPE_JOB_SWITCH:
    case RECORD_TYPE_JOB_REQUEUE:
    case RECORD_TYPE_JOB_CLEAN:
    case RECORD_TYPE_JOB_MOVE:
    case RECORD_TYPE_JOB_MIGRATE:
    case RECORD_TYPE_JOB_MSG:
    case RECORD_TYPE_JOB_SIGNAL:
    case RECORD_TYPE_JOB_SIGACT:
    case RECORD_TYPE_JOB_ACCT:
    case RECORD_TYPE_JOB_FORCE:
    case RECORD_TYPE_JOB_UNFULFILL:
    case RECORD_TYPE_NEW_JGRP:
    case RECORD_TYPE_DEL_JGRP:
    case RECORD_TYPE_MOD_JGRP:
    case RECORD_TYPE_LOG_SWITCH:
    case RECORD_TYPE_JM_JOB_STATUS:
    case RECORD_TYPE_JS_START:
    case RECORD_TYPE_JS_DIE:
    case RECORD_TYPE_QUEUE_CTRL:
    case RECORD_TYPE_HOST_CTRL:
    case RECORD_TYPE_TRAINING_RESULT:
    case RECORD_TYPE_ADDNODE:
    case RECORD_TYPE_DELNODE:
        free_key_value(logrec->numKVs, &(logrec->KVs));
        FREEUP(logrec->KVs);
        logrec->numKVs = 0;

        break;
    default:
        break;
    }

    return;
} // end function free_record

/**
 * @brief       write recode key value into file.
 *
 * @param[in]   log_fp        #1: file to be written
 * @param[in]   log           #2: record
 *
 * @retval      0             #1: succeed
 * @retval      -1            #2: failed
 *
 * @note        file must be opened by caller.
 */
int
write_record(FILE *log_fp, struct record *log)
{
    const char *etype;

    switch (log->type) {
    case RECORD_TYPE_JOB_NEW:
        etype = "JOB_NEW";
        break;
    case RECORD_TYPE_JOB_START:
        etype = "JOB_START";
        break;
    case RECORD_TYPE_JOB_START_ACCEPT:
        etype = "JOB_START_ACCEPT";
        break;
    case RECORD_TYPE_JOB_STATUS:
        etype = "JOB_STATUS";
        break;
    case RECORD_TYPE_JOB_SWITCH:
        etype = "JOB_SWITCH";
        break;
    case RECORD_TYPE_JOB_MOVE:
        etype = "JOB_MOVE";
        break;
    case RECORD_TYPE_QUEUE_CTRL:
        etype = "QUEUE_CTRL";
        break;
    case RECORD_TYPE_HOST_CTRL:
        etype = "HOST_CTRL";
        break;
    case RECORD_TYPE_JS_START:
        etype = "JS_START";
        break;
    case RECORD_TYPE_JS_DIE:
        etype = "JS_DIE";
        break;
    case RECORD_TYPE_JOB_UNFULFILL:
        etype = "UNFULFILL";
        break;
    case RECORD_TYPE_JOB_FINISH:
        etype = "JOB_FINISH";
        break;
    case RECORD_TYPE_JOB_MIGRATE:
        etype = "JOB_MIGRATE";
        break;
    case RECORD_TYPE_PRE_EXEC_START:
        etype = "PRE_EXEC_START";
        break;
    case RECORD_TYPE_JOB_MODIFY:
        etype = "JOB_MODIFY";
        break;
    case RECORD_TYPE_JOB_SIGNAL:
        etype = "JOB_SIGNAL";
        break;
    case RECORD_TYPE_JOB_EXECUTE:
        etype = "JOB_EXECUTE";
        break;
    case RECORD_TYPE_JOB_MSG:
        etype = "JOB_MSG";
        break;
    case RECORD_TYPE_JOB_REQUEUE:
        etype = "JOB_REQUEUE";
        break;
    case RECORD_TYPE_JOB_CLEAN:
        etype = "JOB_CLEAN";
        break;
    case RECORD_TYPE_JOB_SIGACT:
        etype = "JOB_SIGACT";
        break;
    case RECORD_TYPE_JM_JOB_STATUS:
        etype = "JM_JOB_STATUS";
        break;
    case RECORD_TYPE_JOB_ACCT:
        etype = "JOB_ACCT";
        break;
    case RECORD_TYPE_JOB_FORCE:
        etype = "JOB_FORCE";
        break;
    case RECORD_TYPE_LOG_SWITCH:
        etype = "LOG_SWITCH";
        break;
    case RECORD_TYPE_NEW_JGRP:
        etype = "NEW_JGRP";
        break;
    case RECORD_TYPE_DEL_JGRP:
        etype = "DEL_JGRP";
        break;
    case RECORD_TYPE_MOD_JGRP:
        etype = "MOD_JGRP";
        break;
    case RECORD_TYPE_ADDNODE:
        etype = "DYNAMIC_NODE_ADD";
        break;     
    case RECORD_TYPE_DELNODE:
        etype = "DYNAMIC_NODE_DEL";
        break;     
    default:
        quickpool_errno = ERROR_RECORD_TYPE;
        return -1;
    }

    if (fprintf(log_fp, "\"%s\" \"%s\" %ld", etype, log->version, log->log_time) < 0) {

        log_message(ERR, NOCLASS, "%s: fprintf(version and time) failed: %m.", __func__);
        quickpool_errno = ERROR_SYS_CALL;
        return -1;
    }

    int ret=0;
    switch (log->type) {
    case RECORD_TYPE_JOB_NEW:
    case RECORD_TYPE_PRE_EXEC_START:
    case RECORD_TYPE_JOB_START:
    case RECORD_TYPE_JOB_EXECUTE:
    case RECORD_TYPE_JOB_STATUS:
    case RECORD_TYPE_JOB_FINISH:
    case RECORD_TYPE_JOB_MODIFY:
    case RECORD_TYPE_JOB_SWITCH:
    case RECORD_TYPE_JOB_MOVE:
    case RECORD_TYPE_JOB_START_ACCEPT:
    case RECORD_TYPE_JOB_REQUEUE:
    case RECORD_TYPE_JOB_CLEAN:
    case RECORD_TYPE_JOB_MIGRATE:
    case RECORD_TYPE_JOB_MSG:
    case RECORD_TYPE_JOB_SIGNAL:
    case RECORD_TYPE_JOB_SIGACT:
    case RECORD_TYPE_JOB_ACCT:
    case RECORD_TYPE_JOB_FORCE:
    case RECORD_TYPE_JOB_UNFULFILL:
    case RECORD_TYPE_NEW_JGRP:
    case RECORD_TYPE_DEL_JGRP:
    case RECORD_TYPE_MOD_JGRP:
    case RECORD_TYPE_LOG_SWITCH:
    case RECORD_TYPE_JM_JOB_STATUS:
    case RECORD_TYPE_JS_START:
    case RECORD_TYPE_JS_DIE:
    case RECORD_TYPE_QUEUE_CTRL:
    case RECORD_TYPE_HOST_CTRL:
    case RECORD_TYPE_ADDNODE:
    case RECORD_TYPE_DELNODE:
        ret = record_keyvalue(log_fp, log);
        free_record(log);
        break;
    }

    if (fflush(log_fp) != 0) {
        log_message(ERR, NOCLASS, "%s: fflush() failed %m", __func__);
        return -1;
    }

    if (ret == 0) {
        return 0;
    }

    return -1;
} // end function write_record

/**
 * @brief       get a job record from handle for specified job ids.
 *
 * @param[in]   handle        #1: read record from this file handle
 * @param[int]  numjobid      #2: the number of specified job id
 * @param[int]  jobids        #3: specifie job id array
 *
 * @retval      not NULL      #1: succeed
 * @retval      NULL          #2: failed
 *
 * @note        quickpool_errno is set to ERROR_EOF when reading to the end of the file.
 *              file_handle would be changed if reach to end of current file.
 */
struct record *
get_next_record_4job(struct recordFileHandle *handle, unsigned int numjobid, JOBID_T *jobids)
{
    struct record *logRec = NULL;

    while (TRUE) {
        if (NULL != handle->file_handle) {
            logRec = get_jobrecord_fromfile(handle->file_handle, numjobid, jobids);
        } else {
            quickpool_errno = ERROR_EOF;
	}

        if (logRec == NULL && 0 < handle->current_fileno && quickpool_errno == ERROR_EOF
            && handle->current_fileno > handle->last_fileno) {

            if (NULL != handle->file_handle) {
                fclose(handle->file_handle);
            }

            char *strFile = strstr(handle->record_filename, "qp.records");
            if (NULL != strFile) {
                *(strFile-1) = '\0';
            }

            char strRecordFile[PATH_MAX];
            if (1 == handle->current_fileno) {

                snprintf(strRecordFile, PATH_MAX, "%s/qp.records", handle->record_filename);
                handle->current_fileno = 0;
                handle->last_fileno = -1;
            } else {

                snprintf(strRecordFile, PATH_MAX, "%s/qp.records.%d",
                         handle->record_filename, --(handle->current_fileno));
            }

            snprintf(handle->record_filename, PATH_MAX, "%s", strRecordFile);

            FILE *newFileHandle;
            newFileHandle = fopen(strRecordFile, "r");
            if (NULL == newFileHandle) {
                log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "fopen", strRecordFile);
                return NULL;
            }

            handle->file_handle = newFileHandle;

            continue;
        }

        break;
    }

    return logRec;
} // end function get_next_record_4job

void
clean_usage(struct rusage *rusage)
{
    rusage->ru_utime.tv_sec = -1;
    rusage->ru_utime.tv_usec = -1;

    rusage->ru_stime.tv_sec = -1;
    rusage->ru_stime.tv_usec = -1;

    rusage->ru_maxrss = -1;
    rusage->ru_ixrss = -1;

    rusage->ru_isrss = -1;
    rusage->ru_minflt = -1;
    rusage->ru_majflt = -1;
    rusage->ru_nswap = -1;
    rusage->ru_inblock = -1;
    rusage->ru_oublock = -1;

    rusage->ru_idrss = -1;
    rusage->ru_msgsnd = -1;
    rusage->ru_msgrcv = -1;
    rusage->ru_nsignals = -1;
    rusage->ru_nvcsw = -1;
    rusage->ru_nivcsw = -1;
}

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

static int
record_keyvalue(FILE *log_fp, struct record *log)
{
    if (fprintf(log_fp, " %d", log->numKVs) < 0) {
        return ERROR_SYS_CALL;
    }

    unsigned int i;
    for (i=0; i<log->numKVs; i++) {
        if (fprintf(log_fp, " %d", log->KVs[i].key) < 0) {
            return ERROR_SYS_CALL;
        }

        if (add_quotes(log_fp, log->KVs[i].value) < 0) {
            return ERROR_SYS_CALL;
        }
    }

    if (fprintf(log_fp, "\n") < 0) {
        return ERROR_SYS_CALL;
    }

    return 0;
} // end function record_keyvalue

static inline int
copy_quota_string(char **line, char **dest)
{
    char *strNewLine;
    strNewLine = (char *)calloc(strlen(*line)+1, sizeof(char));
    if (NULL == strNewLine) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", strlen(*line)+1);
        return ERROR_MEMORY;
    }

    int ccount;
    if ((ccount = strip_quotes(*line, strNewLine)) < 0) {
        FREEUP(strNewLine);
        return ERROR_FILE_FORMAT;
    }
    *line += ccount + 1;

    *dest = strNewLine;

    return 0;
}

static bool_t
isjobid_in_recordline(char *line, int type, unsigned int num, JOBID_T *jobids)
{
    int     cc;
    JOBID_T jobid = 0;

    unsigned int i;
    int ccount;
    unsigned int numKVs;
    key_value_t *KVs;
    char *jobName;
    switch (type) {
    case RECORD_TYPE_JOB_NEW:
    case RECORD_TYPE_PRE_EXEC_START:
    case RECORD_TYPE_JOB_START:
    case RECORD_TYPE_JOB_STATUS:
    case RECORD_TYPE_JOB_FINISH:    
    case RECORD_TYPE_JOB_EXECUTE:
    case RECORD_TYPE_JOB_MODIFY:
    case RECORD_TYPE_JOB_SWITCH:
    case RECORD_TYPE_JOB_MOVE:
    case RECORD_TYPE_JOB_START_ACCEPT:
    case RECORD_TYPE_JOB_REQUEUE:
    case RECORD_TYPE_JOB_CLEAN:
    case RECORD_TYPE_JOB_MIGRATE:
    case RECORD_TYPE_JOB_MSG:
    case RECORD_TYPE_JOB_SIGNAL:
    case RECORD_TYPE_JOB_SIGACT:
    case RECORD_TYPE_JOB_ACCT:
    case RECORD_TYPE_JOB_FORCE:

        cc = sscanf(line, "%d%n", &numKVs, &ccount);
        if (1 != cc) {
            return FALSE;
        }

        line += ccount + 1;

        KVs = (key_value_t *)calloc(numKVs, sizeof(key_value_t));
        if (NULL == KVs) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", numKVs*sizeof(key_value_t));
            return FALSE;
        }

        for (i=0; i<numKVs; i++) {

            cc = sscanf(line, "%d%n", &(KVs[i].key), &ccount);
            if (1 != cc) {
                log_message(ERR, NOCLASS, "%s: record format is wrong(key[%d]): %s.",
                            __func__, i, line);
                return FALSE;
            }

            line += ccount + 1;
            copy_quota_string(&line, &(KVs[i].value));

            switch (KVs[i].key) {
            case RECORD_KEY_JOBID:
                jobid = atoll(KVs[i].value);
                break;
            case RECORD_KEY_MODIFY_JOBIDS:
                jobName = strstr(KVs[i].value, "[");
                if (jobName != NULL) {
                    *jobName = '\0';
                    jobid = atoll(KVs[i].value);
                    *jobName = '[';
                } else {
                    jobid = atoll(KVs[i].value);
                }
                break;
            default:
                break;
            }

            FREEUP(KVs[i].value);
            if (jobid > 0) {
                break;
            }
        }

        FREEUP(KVs);

        break;
    default:
        return FALSE;
    }

    if (0 == num || type == RECORD_TYPE_JOB_NEW) {
        return TRUE;
    }

    if (0 == jobid) {
        return FALSE;
    }

    for (i=0; i<num; i++) {
        if (ARRAY_JOBID(jobids[i]) == ARRAY_JOBID(jobid)) {
            return TRUE;
        }
    }

    return FALSE;
} // end function isjobid_in_recordline

/**
 * @brief       get job record from the open file.
 *
 * @param[in]   filerecord    #1: the open file handler 
 * @param[int]  numjobid      #2: the number of specified job id
 * @param[int]  jobids        #3: specifie job id array
 *
 * @retval      not NULL      #1: succeed
 * @retval      NULL          #2: failed
 *
 * @note        line format looks like: "type" "version" log_time num_kvs key "value" ...
 *              if specified job ids, read the job record line which job id is in it.
 */
static struct record *
get_jobrecord_fromfile(FILE *filerecord, unsigned int numjobid, JOBID_T *jobids)
{
    static struct record *logRec;

    if (NULL != logRec) {
        free_record(logRec);
    } else {

        logRec = (struct record *)calloc(1, sizeof(struct record));
        if (NULL == logRec) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct record));
            quickpool_errno = ERROR_MEMORY;
            return NULL;
        }
    }

    unsigned int numLine=0;
    while(TRUE) {

        char buf[BUFSIZ];
        char *line;
        line = gf_get_line(filerecord, &numLine, buf);
        if (NULL == line) {

            if (feof(filerecord)) {
                quickpool_errno = ERROR_EOF;
            }

            break;
        }

        int ccount;
        char nameBuf[PATH_MAX];
        if ((ccount = strip_quotes(line, nameBuf)) < 0
            || strlen(nameBuf) >= MAXNAMELEN) {

            quickpool_errno = ERROR_FILE_FORMAT;
            log_message(ERR, NOCLASS, "%s: strip_quotes(%s) failed at line %d.",
                        __func__, line, numLine);
            break;
        }
        line += ccount + 1;

        int kind;
        if ((logRec->type = get_record_type(nameBuf, &kind)) == -1) {
            continue;
        }

        if (kind != RECORD_TYPE_JOB_RELATED) {
            continue;
        }

        if ((ccount = strip_quotes(line, nameBuf)) < 0
            || strlen(nameBuf) >= MAXVERSIONLEN) {

            log_message(ERR, NOCLASS, "%s: strip_quotes(%s) failed at line %d.",
                        __func__, line, numLine);
            quickpool_errno = ERROR_FILE_FORMAT;
            break;
        }

        snprintf(logRec->version, MAXVERSIONLEN, "%s", nameBuf);
        if (atoi(logRec->version) <= 0) {
            log_message(ERR, NOCLASS, "%s: wrong log version(%s) at line %d.",
                        __func__, logRec->version, numLine);
            quickpool_errno = ERROR_FILE_FORMAT;
            break;
        }

        line += ccount + 1;

        int cc, logTime;
        cc = sscanf(line, "%d%n", &logTime, &ccount);
        if (cc != 1) {
            log_message(ERR, NOCLASS, "%s: wrong record format(%s) at line %d.",
                        __func__, line, numLine);
            quickpool_errno = ERROR_FILE_FORMAT;
            break;
        }

        logRec->log_time = logTime;
        line += ccount + 1;

        if (!isjobid_in_recordline(line, logRec->type, numjobid, jobids)) {
            continue;
        }

        quickpool_errno = read_keyvalue_rec(line, logRec);

        break;
    }

    if (quickpool_errno == 0) {
        return logRec;
    }

    return NULL;
} // end function get_jobrecord_fromfile

static int
get_record_type(char *stringtype, int *kind)
{
    int type;

    if (strcmp(stringtype, "JOB_NEW") == 0) {
        type = RECORD_TYPE_JOB_NEW;
    } else if (strcmp(stringtype, "JOB_START") == 0) {
        type = RECORD_TYPE_JOB_START;
    } else if (strcmp(stringtype, "JOB_START_ACCEPT") == 0) {
        type = RECORD_TYPE_JOB_START_ACCEPT;
    } else if (strcmp(stringtype, "JOB_STATUS") == 0) {
        type = RECORD_TYPE_JOB_STATUS;
    } else if (strcmp(stringtype, "JM_JOB_STATUS") == 0) {
        type = RECORD_TYPE_JM_JOB_STATUS;
    } else if (strcmp(stringtype, "JOB_EXECUTE") == 0) {
        type = RECORD_TYPE_JOB_EXECUTE;
    } else if (strcmp(stringtype, "JOB_CLEAN") == 0) {
        type = RECORD_TYPE_JOB_CLEAN;
    } else if (strcmp(stringtype, "JOB_SWITCH") == 0) {
        type = RECORD_TYPE_JOB_SWITCH;
    } else if (strcmp(stringtype, "JOB_MOVE") == 0) {
        type = RECORD_TYPE_JOB_MOVE;
    } else if (strcmp(stringtype, "JOB_FINISH") == 0) {
        type = RECORD_TYPE_JOB_FINISH;
    } else if (strcmp(stringtype, "JOB_MIGRATE") == 0) {
        type = RECORD_TYPE_JOB_MIGRATE;
    } else if (strcmp(stringtype, "PRE_EXEC_START") == 0) {
        type = RECORD_TYPE_PRE_EXEC_START;
    } else if (strcmp(stringtype, "JOB_MODIFY") == 0) {
        type = RECORD_TYPE_JOB_MODIFY;
    } else if (strcmp(stringtype, "QUEUE_CTRL") == 0) {
        type = RECORD_TYPE_QUEUE_CTRL;
    } else if (strcmp(stringtype, "HOST_CTRL") == 0) {
        type = RECORD_TYPE_HOST_CTRL;
    } else if (strcmp(stringtype, "JS_START") == 0) {
        type = RECORD_TYPE_JS_START;
    } else if (strcmp(stringtype, "JS_DIE") == 0) {
        type = RECORD_TYPE_JS_DIE;
    } else if (strcmp(stringtype, "UNFULFILL") == 0) {
        type = RECORD_TYPE_JOB_UNFULFILL;
    } else if (strcmp(stringtype, "JOB_SIGNAL") == 0) {
        type = RECORD_TYPE_JOB_SIGNAL;
    } else if (strcmp(stringtype, "JOB_MSG") == 0) {
        type = RECORD_TYPE_JOB_MSG;
    } else if (strcmp(stringtype, "JOB_REQUEUE") == 0) {
        type = RECORD_TYPE_JOB_REQUEUE;
    } else if (strcmp(stringtype, "JOB_SIGACT") == 0) {
        type = RECORD_TYPE_JOB_SIGACT;
    } else if (strcmp(stringtype, "JOB_ACCT") == 0) {
        type = RECORD_TYPE_JOB_ACCT;
    } else if (strcmp(stringtype, "JOB_FORCE") == 0) {
        type = RECORD_TYPE_JOB_FORCE;
    } else if (strcmp(stringtype, "LOG_SWITCH") == 0) {
        type = RECORD_TYPE_LOG_SWITCH;
    } else if (strcmp(stringtype, "LOG_SWITCH") == 0) {
        type = RECORD_TYPE_LOG_SWITCH;
    } else if (strcmp(stringtype, "NEW_JGRP") == 0) {
        type = RECORD_TYPE_NEW_JGRP;
    } else if (strcmp(stringtype, "DEL_JGRP") == 0) {
        type = RECORD_TYPE_DEL_JGRP;
    } else if (strcmp(stringtype, "MOD_JGRP") == 0) {
        type = RECORD_TYPE_MOD_JGRP;
    } else if (strcmp(stringtype, "TRAINING_RESULT") == 0) {
        type = RECORD_TYPE_TRAINING_RESULT;
    } else if (strcmp(stringtype, "DYNAMIC_NODE_ADD") == 0) {
        type = RECORD_TYPE_ADDNODE;  
    } else if (strcmp(stringtype, "DYNAMIC_NODE_DEL") == 0) {
        type = RECORD_TYPE_DELNODE;  
    } else {
        quickpool_errno = ERROR_FILE_FORMAT;
        *kind = RECORD_TYPE_NON_JOB_RELATED;
        return -1;
    }

    switch (type) {
    case RECORD_TYPE_JOB_NEW:
    case RECORD_TYPE_JOB_START:
    case RECORD_TYPE_JOB_START_ACCEPT:
    case RECORD_TYPE_JOB_STATUS:
    case RECORD_TYPE_JM_JOB_STATUS:
    case RECORD_TYPE_JOB_SWITCH:
    case RECORD_TYPE_JOB_MOVE:
    case RECORD_TYPE_JOB_FINISH:
    case RECORD_TYPE_JOB_MIGRATE:
    case RECORD_TYPE_PRE_EXEC_START:
    case RECORD_TYPE_JOB_MODIFY:
    case RECORD_TYPE_JOB_SIGNAL:
    case RECORD_TYPE_JOB_EXECUTE:
    case RECORD_TYPE_JOB_MSG:
    case RECORD_TYPE_JOB_REQUEUE:
    case RECORD_TYPE_JOB_CLEAN:
    case RECORD_TYPE_JOB_SIGACT:
    case RECORD_TYPE_JOB_ACCT:
    case RECORD_TYPE_JOB_FORCE:
        *kind = RECORD_TYPE_JOB_RELATED;
        break;
    default:
        *kind = RECORD_TYPE_NON_JOB_RELATED;
        break;
    }

    return type;
} // end function get_record_type

static int
read_keyvalue_rec(char *line, struct record *log)
{
    int cc, ccount;
    cc = sscanf(line, "%d%n", &(log->numKVs), &ccount);
    if (cc != 1) {
        log_message(ERR, NOCLASS, "job new record format is wrong(numKVs): %s.", line);
        return ERROR_FILE_FORMAT;
    }

    line += ccount + 1;

    log->KVs = (key_value_t *)calloc(log->numKVs, sizeof(key_value_t));
    if (NULL == log->KVs) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", log->numKVs*sizeof(key_value_t));
        return ERROR_MEMORY;
    }

    unsigned int i;
    for (i=0; i<log->numKVs; i++) {
        cc = sscanf(line, "%d%n", &(log->KVs[i].key), &ccount);
        if (cc != 1) {
            log_message(ERR, NOCLASS, "job new record format is wrong(key[%d]): %s.", i, line);
            return ERROR_FILE_FORMAT;
        }
        line += ccount + 1;

        copy_quota_string(&line, &(log->KVs[i].value));
    }

    return 0;
} // end function read_keyvalue_rec
